package SparkRDD.RDD算子.Transformations.转换操作

import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

class mapPartitionsTest {

  /*
    * mapPartitions 和 map 算子一样，只不过map是针对单条数据进行转换，mapPartitions是针对一整个分区的数据进行转换
    * 所以：
    * map 的 func 参数是单条数据，mapPartitions 的 func 是一个集合(一个分区中的所有数据集)
    * map 的 func 返回值也是单条数据，mapPartitions 的 func 返回值是一个集合
    * mapPartitionsWithIndex 比 mapPartitions 多了个分区参
   */

  // 创建SparkContext
  val conf = new SparkConf().setMaster("local[6]").setAppName("spark_context")
  val sc = new SparkContext(conf)

  @Test
  def mapPartitionsTest1: Unit ={
    // 1.生成数据,设定为2个分区
    val source = sc.parallelize(Seq(1,2,3,4,5,6),2)
    // 2.使用算子
    val process = source.mapPartitions(iter => {
      iter.foreach(item => println(item)) //遍历分区集合数据
      iter     // 返回iter
    })
    // 3.获取结果
    val result = process.collect()
  }

  @Test
  def mapPartitionsTest2: Unit ={
    // 1.生成数据,设定为2个分区
    val source = sc.parallelize(Seq(1,2,3,4,5,6),2)
    // 2.使用算子,将分区里的每条数据*10
    val process = source.mapPartitions(iter => {
      //首先遍历分区里的每条数据,转换完成以后返回新的iter
      //注意这里的iter是scala中的集合概念
      iter.map(item => item*10) // scala中的map返回值就是iter
    })
    // 3.获取结果
    val result = process.collect()
    result.foreach(item => println(item))
  }

  @Test
  def mapPartitionsTest3: Unit ={
    // 1.生成数据,设定为2个分区
    val source = sc.parallelize(Seq(1,2,3,4,5,6),2)
    // 2.使用算子,将分区里的每条数据*10
    val process = source.mapPartitionsWithIndex((index,iter) =>{
      println("index",index)
      iter.foreach(item => println(item))
      iter
    })
    // 3.获取结果
    val result = process.collect()
  }

}
